package cn.ddiancan.xddcloud.psc.mq.service;

import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;

import cn.ddiancan.xddcloud.common.context.SpringContextUtils;
import cn.ddiancan.xddcloud.psc.mq.AbstractMessageHandler;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.MapUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

@Slf4j
@RequiredArgsConstructor
public class XddcloudMessageListener implements MessageListenerConcurrently {

    private final String topic;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        Map<String, AbstractMessageHandler> beanOfType = SpringContextUtils.getBeanOfType(AbstractMessageHandler.class);
        long begin = System.currentTimeMillis();
        AtomicReference<ConsumeConcurrentlyStatus> status =
            new AtomicReference<>(ConsumeConcurrentlyStatus.CONSUME_SUCCESS);
        if (MapUtils.isNotEmpty(beanOfType)) {
            beanOfType.forEach((k, v) -> {
                if (v.getTopic().equalsIgnoreCase(topic)) {
                    try {
                        v.handler(msgs);
                    } catch (Exception e) {
                        status.set(ConsumeConcurrentlyStatus.RECONSUME_LATER);
                        log.error("消息处理失败:{}", e.getMessage(), e);
                    }
                }
            });
        }
        log.info("消息处理完成,耗时:{}", System.currentTimeMillis() - begin);
        return status.get();
    }
}
